package cn.gwm.flink.streaming.task;

import cn.gwm.flink.streaming.beans.BeanSource;
import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.constant.FaultToleranceConstant;
import cn.gwm.flink.streaming.dwd.consumer.EC24Consumer;
import cn.gwm.flink.streaming.dwd.consumer.ES11Consumer;
import cn.gwm.flink.streaming.dwd.consumer.ES13Consumer;
import cn.gwm.flink.streaming.dwd.model.DelayEnum;
import cn.gwm.flink.streaming.dwd.streamsource.CommonStreamSource;
import cn.gwm.flink.streaming.sink.hdfs.HdfsUtil;
import cn.gwm.flink.streaming.sink.kafka.FlinkKafkaUtil;
import cn.gwm.flink.streaming.sink.kafka.KafkaConsumerEnum;
import cn.gwm.utils.ConfigLoader;
import cn.hutool.json.JSONObject;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @ClassName DwdEs11Task
 * @Description dwd 的补充数据 es11 es13 ec24
 * @Author LiangGuang
 * @Date 2023/05/18 14:41
 */
public class DwdBuTask {
    public static void main(String[] args) throws Exception {

        ConfigLoader.init(args);
        String jobName = ConfigLoader.get(FaultToleranceConstant.FLINK_JOB_NAME);

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        BaseTask.envSet(env, jobName);

        DataStream<JSONObject> es11Stream = new ES11Consumer("es11-" + jobName).setTopic("ods_can_status_es11")
                .setGroupId("bulu").setSource(new CommonStreamSource(KafkaConsumerEnum.EARLIEST).setUidSuffix("es11-" + jobName))
                .handle(env);
        DataStream<JSONObject> es13Stream = new ES13Consumer("es13-" + jobName).setTopic("ods_can_status_es13")
                .setGroupId("bulu").setSource(new CommonStreamSource(KafkaConsumerEnum.EARLIEST).setUidSuffix("es13-" + jobName))
                .handle(env);
        DataStream<JSONObject> ec24Stream = new EC24Consumer("ec24-" + jobName).setTopic("ods_can_status_ec24")
                .setGroupId("bulu").setSource(new CommonStreamSource(KafkaConsumerEnum.EARLIEST).setUidSuffix("ec24-" + jobName))
                .handle(env);

        DataStream<JSONObject> dataStream = es11Stream.union(es13Stream, ec24Stream);

        es11Stream.filter(item -> DelayEnum.NORMAL.getCode().equals(item.getStr(BaseFields.DELAY))).uid("dwd-filter-es11-" + jobName)
                .addSink(FlinkKafkaUtil.toKafka("dwd_clean_ev_bu")).uid("dwdToKafka-es11-" + jobName).name("dwdToKafka-dwdbu-es11");
        es13Stream.filter(item -> DelayEnum.NORMAL.getCode().equals(item.getStr(BaseFields.DELAY))).uid("dwd-filter-es13-" + jobName)
                .addSink(FlinkKafkaUtil.toKafka("dwd_clean_ev_bu")).uid("dwdToKafka-es13-" + jobName).name("dwdToKafka-dwdbu-es13");
        ec24Stream.filter(item -> DelayEnum.NORMAL.getCode().equals(item.getStr(BaseFields.DELAY))).uid("dwd-filter-ec24-" + jobName)
                .addSink(FlinkKafkaUtil.toKafka("dwd_clean_ev_bu")).uid("dwdToKafka-ec24-" + jobName).name("dwdToKafka-dwdbu-ec24");

//        dataStream.filter(item -> DelayEnum.NORMAL.getCode().equals(item.getStr(BaseFields.DELAY))).uid("dwd-filter-" + jobName)
//                .addSink(FlinkKafkaUtil.toKafka("dwd_clean_ev_bu")).uid("dwdToKafka-" + jobName).name("dwdToKafka-dwdbu")
//                .setParallelism(env.getParallelism());

        dataStream.addSink(HdfsUtil.hdfsOrcSink("bulu", "dwd/dwdbu", BeanSource.SourceEnum.evFullFieldSource))
                .uid("dwdTdoHdfs-" + jobName).name("dwdTdoHdfs-dwdbu");

        env.execute("dwdEv-dwdbu");


    }
}
